package com.orderstats.consumer;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public class SimpleKafkaConsumer {
    
    private static final String TOPIC = "orders";
    private static final String BOOTSTRAP_SERVERS = "localhost:9092";
    private static final String GROUP_ID = "order-consumer-group";
    private static final String DB_PATH = "/var/lib/tomcat9/webapps/ROOT/orders_system.db";
    
    public static void main(String[] args) {
        System.out.println("🚀 启动简单Kafka消费者...");
        
        // 配置Kafka消费者
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, GROUP_ID);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
        
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList(TOPIC));
        
        System.out.println("📊 开始消费Kafka消息并写入数据库...");
        
        try {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
                
                for (ConsumerRecord<String, String> record : records) {
                    String orderData = record.value();
                    System.out.println("📥 接收到订单数据: " + orderData);
                    
                    // 解析CSV格式数据: category,product,user,quantity,orderTime,rating,isValid
                    String[] fields = orderData.split(",");
                    if (fields.length == 7) {
                        try {
                            insertOrder(fields[0], fields[1], fields[2], 
                                      Integer.parseInt(fields[3]), fields[4], 
                                      Double.parseDouble(fields[5]), fields[6]);
                            System.out.println("✅ 订单数据已写入数据库");
                        } catch (Exception e) {
                            System.err.println("❌ 写入数据库失败: " + e.getMessage());
                        }
                    }
                }
            }
        } catch (Exception e) {
            System.err.println("❌ 消费者异常: " + e.getMessage());
            e.printStackTrace();
        } finally {
            consumer.close();
        }
    }
    
    private static void insertOrder(String category, String product, String userId, 
                                  int quantity, String orderTime, double rating, String isValid) {
        String sql = "INSERT INTO orders (category, product, user_id, quantity, order_time, rating, is_valid) VALUES (?, ?, ?, ?, ?, ?, ?)";
        
        try (Connection conn = DriverManager.getConnection("jdbc:sqlite:" + DB_PATH);
             PreparedStatement pstmt = conn.prepareStatement(sql)) {
            
            pstmt.setString(1, category);
            pstmt.setString(2, product);
            pstmt.setString(3, userId);
            pstmt.setInt(4, quantity);
            pstmt.setString(5, orderTime);
            pstmt.setDouble(6, rating);
            pstmt.setString(7, isValid);
            
            pstmt.executeUpdate();
            
        } catch (SQLException e) {
            System.err.println("❌ 数据库操作失败: " + e.getMessage());
            throw new RuntimeException(e);
        }
    }
} 